package com.haiyou.data.common.persister.mysql.persister;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.util.collect.CollectionUtils;
import com.haiyou.common.util.mysql.BatchUtils;
import com.haiyou.data.common.persister.mysql.help.DatabaseHelper;
import com.haiyou.data.common.persister.mysql.help.EntityHelper;
import com.haiyou.data.common.persister.mysql.help.SqlHelper;
import com.haiyou.data.common.persister.mysql.help.TableHelper;
import com.haiyou.data.common.persister.mysql.meta.ColumnMeta;
import com.haiyou.data.common.persister.mysql.annotation.Table.RollPolicy;

import lombok.extern.slf4j.Slf4j;

/**
 * 使用预编译sql语句进行批量操作mysql的持久器
 * 
 * @author Administrator
 *
 * @param <Id>
 * @param <E>
 */
@Slf4j
public abstract class PreparedPersister<Id extends Serializable & Comparable<Id>, E extends Entity<Id>>
		extends Persister<Id, E> {

	protected class MyMap<K, V> extends HashMap<K, V> {

		/**
		 * 
		 */
		private static final long serialVersionUID = 2898937650950899875L;

		public List<V> getAll(Iterable<?> keys) {
			List<V> list = Lists.newArrayListWithCapacity(Iterables.size(keys));
			keys.forEach(k -> list.add(get(k)));
			return list;
		}
	}

	protected MyMap<Id, E> toMap(Iterable<? extends E> iterable) {
		MyMap<Id, E> map = new MyMap<>();
		iterable.forEach(e -> {
			if(e.getId() == null) {
				//创建了过多的char[] 修改一下
				Objects.requireNonNull(e.getId(),"主键为null: "+ e);
			}
			map.put(e.getId(), e);
		});
		return map;
	}

	@Override
	public void insert(Iterable<? extends E> iterable) {

		checkTableIfDoesExist();
		
		if (tableMeta.getRollPolicy() != RollPolicy.NONE || TableHelper.pkAutoIncr(tableMeta)) {
			String db = DatabaseHelper.getOnlyReallyName(databaseMeta);
			String tab = RollPolicy.fixName(tableMeta.getSimpleName(), tableMeta.getRollPolicy());
			insert(db, tab, Lists.newArrayList(iterable));
			return;
		}

		MyMap<Id, E> map = toMap(iterable);
		
		DatabaseHelper.matchedDatabases(databaseMeta, map.keySet())
				.forEach((db, dbids) -> TableHelper.matchedTables(tableMeta, dbids).forEach((tab, tabids) -> {
					insert(db, tab, map.getAll(tabids));
				}));
	}

	private void insert(String db, String tab, List<E> values) {
		int size = values.size();
		if (size <= sqlBatchNum()) {
			batchInsert(db, tab, values);
		} else {
			BatchUtils.toLists(values, sqlBatchNum()).forEach(list -> batchInsert(db, tab, list));
		}
	}

	private void batchInsert(String db, String tab, List<E> values) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = on(db);
			preparedStatement = connection.prepareStatement(
					sql = SqlHelper.preparedInsert(entityClass, db, tab, !TableHelper.pkAutoIncr(tableMeta)));
			for (Entity<Id> entity : values) {
				int parameterIndex = 1;
				for (int i = !TableHelper.pkAutoIncr(tableMeta) ? 0 : 1; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
			else
				log.debug("[{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
		}
	}

	@Override
	public void insertOrUpdate(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();
		//列表转换map
		MyMap<Id, E> map = toMap(iterable);
		DatabaseHelper.matchedDatabases(databaseMeta, map.keySet())
				.forEach((db, dbids) -> {
			TableHelper.matchedTables(tableMeta, dbids)
				.forEach((tab, tabids) -> {
				List<E> values = map.getAll(tabids);
				int size = values.size();
				if (size <= sqlBatchNum()) {
					batchInsertOrUpdate(db, tab, values);
				} else {
					BatchUtils.toLists(values, sqlBatchNum()).forEach(list -> batchInsertOrUpdate(db, tab, list));
				}
			});
		});
	}

	protected void batchInsertOrUpdate(String db, String tab, List<E> values) {
		
		checkTableIfDoesExist();
		
		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = on(db);
			preparedStatement = connection
					.prepareStatement(sql = SqlHelper.preparedInsertOnDuplicateKeyUpdate(entityClass, db, tab));
			
			for (Entity<Id> entity : values) {
				int parameterIndex = 1;
				for (int i = 0; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				for (int i = 0; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly()) {
				log.warn("SLOWLY: [{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
			}
			else {
				if(log.isDebugEnabled()) {
					log.debug("[{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
				}
			}
		}
	}

	@Override
	public void insertIfNotExists(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		MyMap<Id, E> map = toMap(iterable);
		DatabaseHelper.matchedDatabases(databaseMeta, map.keySet()).forEach((db, dbids) -> {
			TableHelper.matchedTables(tableMeta, dbids).forEach((tab, tabids) -> {
				List<E> values = map.getAll(tabids);
				int size = values.size();
				if (size <= sqlBatchNum()) {
					batchInsertIfNotExists(db, tab, values);
				} else {
					BatchUtils.toLists(values, sqlBatchNum()).forEach(list -> batchInsertIfNotExists(db, tab, list));
				}
			});
		});
	}

	private void batchInsertIfNotExists(String db, String tab, List<E> values) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = on(db);
			preparedStatement = connection
					.prepareStatement(sql = SqlHelper.preparedInsertIfNotExists(entityClass, db, tab));
			for (Entity<Id> entity : values) {
				int parameterIndex = 1;
				for (int i = 0; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
			else
				log.debug("[{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
		}
	}

	@Override
	public void replace(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		MyMap<Id, E> map = toMap(iterable);
		DatabaseHelper.matchedDatabases(databaseMeta, map.keySet()).forEach((db, dbids) -> {
			TableHelper.matchedTables(tableMeta, dbids).forEach((tab, tabids) -> {
				List<E> values = map.getAll(tabids);
				int size = values.size();
				if (size <= sqlBatchNum()) {
					batchReplace(db, tab, values);
				} else {
					BatchUtils.toLists(values, sqlBatchNum()).forEach(list -> batchReplace(db, tab, list));
				}
			});
		});
	}

	protected void batchReplace(String db, String tab, List<E> values) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = on(db);
			preparedStatement = connection.prepareStatement(sql = SqlHelper.preparedReplace(entityClass, db, tab));
			for (Entity<Id> entity : values) {
				int parameterIndex = 1;
				for (int i = 0; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
			else
				log.debug("[{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
		}
	}

	@Override
	public void update(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		MyMap<Id, E> map = toMap(iterable);
		DatabaseHelper.matchedDatabases(databaseMeta, map.keySet()).forEach((db, dbids) -> {
			TableHelper.matchedTables(tableMeta, dbids).forEach((tab, tabids) -> {
				List<E> values = map.getAll(tabids);
				int size = values.size();
				if (size <= sqlBatchNum()) {
					batchUpdate(db, tab, values);
				} else {
					BatchUtils.toLists(values, sqlBatchNum()).forEach(list -> batchUpdate(db, tab, list));
				}
			});
		});
	}

	protected void batchUpdate(String db, String tab, List<E> values) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = on(db);
			preparedStatement = connection.prepareStatement(sql = SqlHelper.preparedUpdate(entityClass, db, tab));
			for (Entity<Id> entity : values) {
				int parameterIndex = 1;
				for (int i = 1; i < tableMeta.getColumnMetas().size(); i++) {
					ColumnMeta columnMeta = tableMeta.getColumnMetas().get(i);
					Object x = cast(columnMeta, entity);
					preparedStatement.setObject(parameterIndex++, x);
				}
				Object x = cast(tableMeta.getIdColumnMeta(), entity);
				preparedStatement.setObject(parameterIndex++, x);
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
			else
				log.debug("[{}] elapsed [{} ms],size={}", sql, elapsed, values.size());
		}
	}

	
	/**
	 * 不限制查询
	 * @return
	 */
	public List<E> queryAllNotLimit() {
		List<Map<String, Object>> list = Lists.newArrayList();
		
		for (String db : databaseMeta.getReallyNames()) {
			
			for (String tab : TableHelper.allTables(tableMeta)) {
				list.addAll(executeQueryAll(db, tab));
			}
		}
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		
		return result;
	}
	
	public List<Map<String, Object>> executeQueryAll(String db, String tab) {
		return executeQuery(db, String.format("select *from `%s`.`%s` ", db, tab));
	}
	
	
	
}
